package com.zeratul.common.event.core.handler.redis.config;

import com.zeratul.common.event.core.apply.subscribe.IZtEventHandler;
import com.zeratul.common.event.core.handler.local.config.SpringEventHandlerAutoConfiguration;
import com.zeratul.common.event.core.handler.redis.RedisPublishHandler;
import com.zeratul.common.event.core.handler.redis.RedisSubscribeHandler;
import com.zeratul.common.event.core.server.publish.inter.DelayMessageProducerInterceptor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.RedisSerializer;

import java.util.List;

@Slf4j
@AutoConfiguration(before = SpringEventHandlerAutoConfiguration.class)
@ConditionalOnClass(RedisTemplate.class)
@ConditionalOnProperty(prefix = "zt-event", value = "type", havingValue = "redis")
@Import(DelayMessageProducerInterceptor.class)
public class RedisEventAutoConfiguration {

    /**
     * 用于发布与订阅的RedisTemplate
     */
    @Bean
    public RedisTemplate<String, byte[]> eventRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, byte[]> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        template.setKeySerializer(RedisSerializer.string());
        template.setValueSerializer(RedisSerializer.byteArray());
        template.setHashKeySerializer(RedisSerializer.string());
        template.setHashValueSerializer(RedisSerializer.byteArray());
        return template;
    }

    /**
     * Redis发布者
     */
    @Bean
    public RedisPublishHandler redisPublishHandler(){
        log.debug("【事件总线】基于Redis的事件发布者加载...");
        return new RedisPublishHandler();
    }

    /**
     * Redis订阅者
     */
    @Bean
    @ConditionalOnBean(IZtEventHandler.class)
    public RedisSubscribeHandler redisSubscribeHandler(){
        log.debug("【事件总线】基于Redis的事件订阅者加载...");
        return new RedisSubscribeHandler();
    }


    /**
     * Redis消费者
     * @param connectionFactory
     * @return
     */
    @Bean
    @ConditionalOnBean(RedisSubscribeHandler.class)
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, RedisSubscribeHandler redisSubscribeHandler, List<String> getTopics) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);

        //创建Redis订阅者主题集合
        List<PatternTopic> patternTopics = getTopics.stream().map(PatternTopic::new).toList();
        container.addMessageListener(new MessageListenerAdapter(redisSubscribeHandler), patternTopics);
        return container;
    }
}
